package com.foreveross.fetl.common.hbase.client;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;

import com.foreveross.fetl.common.db.hbase.HbaseDriverManager;

public class HBaseHelper {

	public static HbaseDriverManager hbaseDriverManager = null;
	static{
		hbaseDriverManager = HbaseDriverManager.getInstance();
	}
	/**
	 * 新增一条记录
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param family
	 * @param qualifier
	 * @param value
	 * @throws Exception
	 */
	public static void addRecord(String tableName, String rowKey,
			String family, String qualifier, String value) throws IOException {
		HTableInterface table = null;
		Put put = null;
		try{
			put = new Put(Bytes.toBytes(rowKey));
			put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier),
					Bytes.toBytes(value));
			table = hbaseDriverManager.getHtable(tableName);
			table.put(put);
			//System.out.println("insert recored " + new String(put.getRow()) + " to table " +tableName +" ok.");
		}finally{
			if(table != null) hbaseDriverManager.releaseHTable(table);
			table = null;
			put = null;
		}

		
	}

	/**
	 * 根据rowkey删除一条记录
	 * 
	 * @param tableName
	 * @param rowKey
	 * @throws IOException
	 */
	public static void delRecord(String tableName, String rowKey)
			throws IOException {
		HTableInterface table = null;
		try{
			table = hbaseDriverManager.getHtable(tableName);
			table.delete(new Delete(rowKey.getBytes()));
		}finally{
			if(table != null) hbaseDriverManager.releaseHTable(table);
			table = null;
		}
	}
	
	/**
	 * 清空表
	 * @param tableName
	 * @throws IOException
	 */
	public static void delAllRecord(String tableName) throws IOException{
		HTableInterface table = null;
		ResultScanner ss = null;
		try{
			table = hbaseDriverManager.getHtable(tableName);
			Scan s = new Scan();
			ss = table.getScanner(s);
			for (Result r : ss) {
				for (KeyValue kv : r.raw()) {
					table.delete(new Delete(kv.getRow()));
					
				}
			}
			//table.delete(new Delete(rowKey.getBytes()));
		}finally{
			if(table != null) hbaseDriverManager.releaseHTable(table);
			table = null;
		}
	}

	/**
	 * 通过一个rowkey取得对应的记录
	 * 
	 * @param tableName
	 *            ：表名
	 * @param rowKey
	 * @return
	 * @throws IOException
	 */
	public static Result getOneRecord(String tableName, String rowKey)
			throws IOException {
		Get get = null;
		Result rs = null;
		HTableInterface table = null;
		try{
			get = new Get(rowKey.getBytes("utf-8"));
			table = hbaseDriverManager.getHtable(tableName);
			rs = table.get(get);
		}finally{
			if(table != null) hbaseDriverManager.releaseHTable(table);
			table = null;
			get = null;
		}
		return rs;
	}

	/**
	 * 统计某时段内hbase库中的数据量
	 * 
	 * @param tableName
	 * @param minStamp
	 * @param maxStamp
	 * @return
	 * @throws IOException
	 */
	public static ResultScanner getRecordsByTimestamp(String tableName,
			long minStamp, long maxStamp) throws IOException {
		Scan s = null;
		HTableInterface table = null;
		ResultScanner ss = null;
		try{
			 s = new Scan();
			 table = hbaseDriverManager.getHtable(tableName);
			 s.setTimeRange(minStamp, maxStamp);
			 ss = table.getScanner(s);
		}finally{
			hbaseDriverManager.releaseHTable(table);
			table = null;
			s = null;
		}
		return ss;
	}
	
	/**
	 * 统计某时段内hbase库中的数据量
	 * 
	 * @param tableName
	 * @param minStamp
	 * @param maxStamp
	 * @param family
	 * @return
	 * @throws IOException
	 */
	public static ResultScanner getRecordsByTimestamp(String tableName,
			long minStamp, long maxStamp,String[] family) throws IOException {
		Scan s = null;
		HTableInterface table = null;
		ResultScanner ss = null;
		try{
			 s = new Scan();
			 table = hbaseDriverManager.getHtable(tableName);
			 if (family != null && family.length > 0) {
					for (String cf : family) {
						s.addFamily(cf.getBytes());
					}
				}
			
			 s.setTimeRange(minStamp, maxStamp);
			 ss = table.getScanner(s);
		}finally{
			hbaseDriverManager.releaseHTable(table);
			table = null;
			s = null;
		}
		return ss;
	}

	/**
	 * 
	 * 根据条件查询hbase数据库
	 * 
	 * @param tableName
	 *            :表名
	 * @param startRowKey
	 *            ：开始rowkey
	 * @param stopRowKey
	 *            ：结束rowkey
	 * @param minStamp 
	 * @param maxStamp
	 * @return
	 * @throws IOException
	 */
	public static ResultScanner getRecords(String tableName,
			String startRowKey, String stopRowKey,long minStamp, long maxStamp) throws IOException {
		ResultScanner ss = null;
		Scan s = null;
		HTableInterface table = null;
		try{
			 s = new Scan();
			 s.setCaching(100);
			 table = hbaseDriverManager.getHtable(tableName);
			if (StringUtils.isNotBlank(startRowKey)) {
				s.setStartRow(startRowKey.getBytes());
			}
			if (StringUtils.isNotBlank(stopRowKey)) {
				s.setStopRow(stopRowKey.getBytes());
			}
		
			s.setTimeRange(minStamp, maxStamp);
			ss = table.getScanner(s);
		}finally{
			hbaseDriverManager.releaseHTable(table);
			table = null;
			s = null;
		}
		
		return ss;
	}
	
	

	/**
	 * 根据条件查询hbase数据库
	 * 
	 * @param tableName
	 *            :表名
	 * @param startRowKey
	 *            ：开始rowkey
	 * @param stopRowKey
	 *            ：结束rowkey
	 * @param filters
	 *            ：过滤器
	 * @return
	 */
	public static ResultScanner getRecords(String tableName,
			String startRowKey, String stopRowKey) throws IOException {
		return HBaseHelper.getRecords(tableName, startRowKey, stopRowKey, null,
				null);
	}

	/**
	 * 根据条件查询hbase数据库
	 * 
	 * @param tableName
	 *            :表名
	 * @param startRowKey
	 *            ：开始rowkey
	 * @param stopRowKey
	 *            ：结束rowkey
	 * @param family
	 *            ：列簇
	 * @param filters
	 *            ：过滤器
	 * @return
	 */
	public static ResultScanner getRecords(String tableName,
			String startRowKey, String stopRowKey, String[] family,
			List<Filter> filters) throws IOException {
		ResultScanner ss = null;
		Scan s = null;
		HTableInterface table = null;
		try{
			 s = new Scan();
			 s.setCaching(100);
			 table = hbaseDriverManager.getHtable(tableName);
			if (StringUtils.isNotBlank(startRowKey)) {
				s.setStartRow(startRowKey.getBytes());
			}
			if (StringUtils.isNotBlank(stopRowKey)) {
				s.setStopRow(stopRowKey.getBytes());
			}
		
			if (family != null && family.length > 0) {
				for (String cf : family) {
					s.addFamily(cf.getBytes());
				}
			}
			if (filters != null && filters.size() > 0) {
				FilterList list = new FilterList(filters);
				s.setFilter(list);
			}
			ss = table.getScanner(s);
		}finally{
			hbaseDriverManager.releaseHTable(table);
			table = null;
			s = null;
		}
		return ss;
	}
	
	/**
	 * 根据条件查询hbase数据库
	 * @param tableName 表名
	 * @param startRowKey 开始rowkey
	 * @param stopRowKey 结束rowkey
	 * @param columns    {planeinfo:fromCity,planeinfo:toCity} 列
	 * @return
	 * @throws IOException
	 */
	public static ResultScanner getRecords(String tableName,
			String startRowKey, String stopRowKey, String[] columns) throws IOException {
		ResultScanner ss = null;
		Scan s = null;
		HTableInterface table = null;
		try{
			 s = new Scan();
			 s.setCaching(100);
			 table = hbaseDriverManager.getHtable(tableName);
			if (StringUtils.isNotBlank(startRowKey)) {
				s.setStartRow(startRowKey.getBytes());
			}
			if (StringUtils.isNotBlank(stopRowKey)) {
				s.setStopRow(stopRowKey.getBytes());
			}
			if (StringUtils.isNotBlank(stopRowKey)) {
				s.setStopRow(stopRowKey.getBytes());
			}
			if (columns != null && columns.length > 0) {
				for (String column : columns) {
					String[] cs=column.split(":");
					if(null==cs || cs.length==0)
						continue;
					s.addColumn(Bytes.toBytes(cs[0]), Bytes.toBytes(cs[1]));
				}
			}
			ss = table.getScanner(s);
		}finally{
			hbaseDriverManager.releaseHTable(table);
			table = null;
			s = null;
		}
		return ss;
	}
	
	/**
	 * 添加一行记录
	 * 
	 * @param tableName
	 * @param puts
	 * @throws Exception
	 */
	public static void addRecord(String tableName, Put put) throws Exception {
		HTableInterface table = null;
		try {
			table = hbaseDriverManager.getHtable(tableName);
			table.put(put);
			//System.out.println("insert recored " + new String(put.getRow()) + " to table " +tableName +" ok.");
		} finally{
			hbaseDriverManager.releaseHTable(table); 
			table = null;
		}
	}
	public static ResultScanner getAllRecord(String tableName) {
		ResultScanner ss = null;
		try {
			Scan s = new Scan();
			HTableInterface table = hbaseDriverManager.getHtable(tableName);
			ss = table.getScanner(s);
			for (Result r : ss) {
				for (KeyValue kv : r.raw()) {
					System.out.print(new String(kv.getRow()) + " ");
					System.out.print(new String(kv.getFamily()) + ":");
					System.out.print(new String(kv.getQualifier()) + " ");
					// System.out.print(kv.getTimestamp() + " ");
					System.out.println(new String(kv.getValue()));
				}
			}
			hbaseDriverManager.releaseHTable(table);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return ss;
	}
	public static void main(String[] args) {
		ResultScanner rs = getAllRecord("LP_FREIGHT_MONITOR");
		printResult(rs);
	}
	public static void printResult(ResultScanner scan) {
		for (Result r : scan) {
			for (KeyValue kv : r.raw()) {
				System.out.print("   rowKey: " + new String(kv.getRow()));
				System.out.print("   family: " + new String(kv.getFamily()));
				System.out.print("   column: " + new String(kv.getQualifier())
						+ " ");
				System.out.print("    value: " + new String(kv.getValue()));
				System.out.println("   timstamp: " + kv.getTimestamp() + " ");
			}
		}
	}
	/**
	 * 添加多行记录
	 * 
	 * @param tableName
	 * @param puts
	 * @throws Exception
	 */
	public static void addRecord(String tableName, List<Put> puts)
			throws Exception {
		HTableInterface table = null;
		try {
		    table = hbaseDriverManager.getHtable(tableName);
			table.put(puts);
		} finally{
			hbaseDriverManager.releaseHTable(table);
			table = null;
		}
	}
	/**
	 * 将hbase中的一行记录转换为方便统计的Map结构
	 * 
	 * @param res
	 *            Map<列簇，Map<列名，列值>>
	 * @return
	 */
	public static Map<String, Map<String, String>> transformHBaseRowToMap(
			Result res) {
		Map<String, Map<String, String>> temp = new HashMap<String, Map<String, String>>();// 保存HBASE中存储的信息
//	for (Cell ce :res.rawCells()) {
//		KeyValue kv=new KeyValue(ce);
//		CellUtil.cloneFamily(ce);
//			try {
//				if (temp.get(new String(kv.getFamily())) != null) {
//					temp.get(new String(kv.getFamily())).put(
//							new String(kv.getQualifier()),
//							new String(kv.getValue()));
//				} else {
//					Map<String, String> colVal = new HashMap<String, String>();
//					colVal.put(new String(kv.getQualifier()),
//							new String(kv.getValue()));
//					temp.put(new String(kv.getFamily()), colVal);
//				}
//			} catch (Exception e) {
//				e.printStackTrace();
//			}
//		}
	
	for (Cell ce :res.rawCells()) {
			try {
				if (temp.get(new String(CellUtil.cloneFamily(ce))) != null) {
					temp.get(new String(CellUtil.cloneFamily(ce))).put(
							new String(CellUtil.cloneQualifier(ce)),
							new String(CellUtil.cloneValue(ce)));
				} else {
					Map<String, String> colVal = new HashMap<String, String>();
					colVal.put(new String(CellUtil.cloneQualifier(ce)),
							new String(CellUtil.cloneValue(ce)));
					temp.put(new String(CellUtil.cloneFamily(ce)), colVal);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
		return temp;
	}

	
	
}
